package dim


import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.{Cell, HBaseConfiguration, HConstants, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Result, ResultScanner, Scan, Table}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.flink.configuration
import java.util
import scala.collection.JavaConverters._
/**
 * @author: yehw
 * @date: 2021/3/12 下午7:32
 * @description: 从HBase中读取lagou_area
 */
class HBaseReader extends RichSourceFunction[(String, String)] {
  private var conn: Connection = null
  private var table: Table = null;
  private var scan: Scan = null;

  override def open(parameters: configuration.Configuration): Unit = {
    val conf: Configuration = HBaseConfiguration.create()
    conf.set(HConstants.ZOOKEEPER_QUORUM, "linux121,linux123")
    conf.set(HConstants.ZOOKEEPER_CLIENT_PORT, "2181")
    conf.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, 30000)
    conf.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, 30000)
    val tableName: TableName = TableName.valueOf("lagou_area")
    val cf1: String = "f1"
    conn = ConnectionFactory.createConnection(conf)
    table = conn.getTable(tableName)
    scan = new Scan()
    scan.addFamily(Bytes.toBytes(cf1))
  }

  override def run(ctx: SourceFunction.SourceContext[(String, String)]): Unit = {
    val rs: ResultScanner = table.getScanner(scan)
    val iterator: util.Iterator[Result] = rs.iterator()
    while (iterator.hasNext) {
      val result: Result = iterator.next()
      val rowKey: String = Bytes.toString(result.getRow)
      val buffer: StringBuffer = new StringBuffer()
      for (cell: Cell <- result.listCells().asScala) {
        val value: String = Bytes.toString(cell.getValueArray, cell.getValueOffset, cell.getValueLength)
        buffer.append(value).append("-")
      }
      val valueString: String = buffer.replace(buffer.length() - 1, buffer.length(), "").toString
      ctx.collect((rowKey, valueString))
    }
  }

  override def cancel(): Unit = ???
}
